package com.xzy.rabbit.demo.publisherconfirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;

/**
 * 批量确认
 *
 * @author xzy
 * @date 2021年9月26日
 */
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
class BatchACK {

    public static long workTime;

    /**
     * 每发布多少条消息进行一次确认
     */
    private static final int BATCH_ACK_COUNT = 10;

    /**
     * 同步确认 + 批量确认
     */
    public static void produce(Connection connection, String queueName, int messageCount, CountDownLatch sendMsg) throws IOException, TimeoutException, InterruptedException {
        // 1.获取信道
        try (Channel channel = connection.createChannel()) {

            // 2.声明队列
            channel.queueDeclare(queueName, false, false, false, null);

            // 3.开启发布确认（！！！）
            channel.confirmSelect();

            // 3.发送消息
            long begin = System.currentTimeMillis();
            int batchMsgCount = 0;
            for (int i = 0; i < messageCount; i++) {
                String message = "msg" + i;
                channel.basicPublish("", queueName, null, message.getBytes());
                log.info("发布：{}", message);
                batchMsgCount++;

                // 3.1批量确认
                if (batchMsgCount == BATCH_ACK_COUNT) {
                    if (channel.waitForConfirms()) {
                        log.info("确认");
                    } else {
                        // 批量确认无法精确定位发布失败的消息
                        log.info("不确认");
                    }
                    batchMsgCount = 0;
                }
            }
            workTime = System.currentTimeMillis() - begin;
            log.info("总耗时：{}", workTime);
            sendMsg.countDown();
        }

    }
}
